package com.hzg.stream.state;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

/**
 * @Author: huangzhigao
 * @Date: 2022/4/20 15:58
 */
public class ValueStateFunction extends RichMapFunction<String, String> {
    private ValueState<String> userValueState;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //注册stateDescriptor
        ValueStateDescriptor<String> valueState = new ValueStateDescriptor<>("valueState", String.class);
        userValueState = getRuntimeContext().getState(valueState);
    }

    @Override
    public String map(String value) throws Exception {
        String stateValue = userValueState.value();
        userValueState.update(value + "-" + stateValue);
        return value + "-" + stateValue;
    }
}
